package groupOne.app.DIM;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.configTable;
import groupOne.common.Constant;
import groupOne.util.DruidDSUtil;
import groupOne.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;

import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;

/**
 * @Author: LixuanFeng
 * @Date: 2022/8/31 9:36
 * @From: GoodGoodStudyDayDayUp
 */

public class DIM_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {
        new DIM_lixuanfeng().init(2010,2,"DIM", Constant.TOPIC_ODS_DB);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 对业务数据做过滤 ETL
        SingleOutputStreamOperator<JSONObject> etlStream = etlData(stream);

        //2. 通过读取配置表 获得配置流
        SingleOutputStreamOperator<configTable> configStream = readConfigTable(env);

        //3.在phoenix中建表，并让两个流 connect 到一起
        SingleOutputStreamOperator<Tuple2<JSONObject, configTable>> connectStream = createTableAndConnect(etlStream, configStream);

        //4.过滤掉主流数据中有，但是配置表中却不需要的字段
        SingleOutputStreamOperator<Tuple2<JSONObject, configTable>> filtedStream = filteNoUseColum(connectStream);

        //5.数据写入到Hbase中
        writeToHbase(filtedStream);

    }

    private void writeToHbase(SingleOutputStreamOperator<Tuple2<JSONObject, configTable>> filtedStream) {



        filtedStream
                .addSink(new RichSinkFunction<Tuple2<JSONObject, configTable>>() {

                    private DruidDataSource druidDataSource;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 获取连接池
                        druidDataSource = DruidDSUtil.getDruidDataSource();
                    }

                    @Override
                    public void close() throws Exception {
                        druidDataSource.close();  // app关闭之前关闭连接池
                    }

                    @Override
                    public void invoke(Tuple2<JSONObject, configTable> value, Context context) throws Exception {

                        // 每来一条数据, 从连接池获取一个可用的连接, 这样可以避免长连接被服务器自动关闭的问题.
                        DruidPooledConnection conn = druidDataSource.getConnection();

                        JSONObject data = value.f0;
                        configTable config = value.f1;

                        // 1. 拼接sql语句. 一定有占位符
                        // upsert into user(aa,b,c)values(?,?,?)
                        StringBuilder sql = new StringBuilder();
                        sql
                                .append("upsert into ")
                                .append(config.getSinkTable())
                                .append("(")
                                .append(config.getSinkColumns())
                                .append(")values(")
                                .append(config.getSinkColumns().replaceAll("[^,]+", "?"))
                                .append(")");
                        System.out.println("插入语句: " + sql);

                        PreparedStatement ps = conn.prepareStatement(sql.toString());

                        // 3. 给占位符赋值 TODO
                        String[] cs = config.getSinkColumns().split(",");
                        for (int i = 0, count = cs.length; i < count; i++) {
                            Object v = data.get(cs[i]);  // null + ""="null"
                            String vv = v == null ? null : v.toString();
                            ps.setString(i + 1, vv);
                        }

                        // 4. 执行sql
                        ps.execute();
                        conn.commit();
                        // 5.关闭PrepareStatement
                        ps.close();
                        // 6. 归还连接
                        conn.close();
                    }
                });

    }


    private SingleOutputStreamOperator<Tuple2<JSONObject, configTable>> filteNoUseColum(SingleOutputStreamOperator<Tuple2<JSONObject, configTable>> connectStream) {

      return connectStream.map(new MapFunction<Tuple2<JSONObject, configTable>, Tuple2<JSONObject, configTable>>() {
            @Override
            public Tuple2<JSONObject, configTable> map(Tuple2<JSONObject, configTable> value) throws Exception {

                //取出数据中的字段
                JSONObject data = value.f0;

                //取出配置表中的字段
                String[] configColumns = value.f1.getSinkColumns().split(",");

                List<String> columslist = Arrays.asList(configColumns);

                //过滤掉数据在配置中没有的字段key
                data.keySet().removeIf(key -> !columslist.contains(key) && !"op_type".equals(key));

                return value;
            }
        });

    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, configTable>> createTableAndConnect(SingleOutputStreamOperator<JSONObject> etlStream, SingleOutputStreamOperator<configTable> configStream) {
        //第一步：先建表
            //1.获取phoenix连接, 一般在open方法里获取，减少连接次数
            //2.拼接一个SQL语句
            //3.获取一个预编译对象
            //4.用连接调用预编译对象得到 预处理语句

        // 在Phoenix建表. 建表语句应该在广播之前, 否则会出现多次建表的情况
        configStream
                .map(new RichMapFunction<configTable, configTable>() {

                    private Connection connection;

                    @Override
            public void open(Configuration parameters) throws Exception {
                        connection = JdbcUtil.getPhoenixConnection();
            }

            @Override
            public void close() throws Exception {
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public configTable map(configTable value) throws Exception {

                StringBuffer SQL = new StringBuffer();

                SQL
                        .append("create table if not exists ")
                        .append(value.getSinkTable())
                        .append("(")
                        .append(value.getSinkColumns().replaceAll("[^,]+", "$0 varchar"))
                        .append(", constraint pk primary key(")
                        .append(value.getSinkPk() == null ? "id" : value.getSinkPk())
                        .append("))")
                        .append(value.getSinkExtend() == null ? "" : value.getSinkExtend());

                System.out.println("phoenix建表语句: " + SQL);

                // 2. 获取预处理语句
                PreparedStatement ps = connection.prepareStatement(SQL.toString());

                //执行  关闭
                ps.execute();
                ps.close();
                return value;
            }
        });

        //第二步 把两个流connect，先把配置流当成广播流
         //1.先new一个描述器
        MapStateDescriptor<String, configTable> broadcastDescriptor = new MapStateDescriptor<>("broadcast", String.class, configTable.class);

        //2.把配置流 得到广播流
        BroadcastStream<configTable> configTableBroadcastStream = configStream.broadcast(broadcastDescriptor);

        //把配置流作用到流上 connect

        return
        etlStream
                .connect(configTableBroadcastStream)
                .process(new BroadcastProcessFunction<JSONObject, configTable, Tuple2<JSONObject,configTable>>() {


                    //把配置信息先写入状态，
                    @Override
                    public void processBroadcastElement(configTable config, Context ctx, Collector<Tuple2<JSONObject, configTable>> out) throws Exception {
                        // key: source_table
                        // value: TableProcess
                        BroadcastState<String, configTable> broadcastState = ctx.getBroadcastState(broadcastDescriptor);

                        String key = config.getSourceTable();

                        broadcastState.put(key,config);

                    }


                    @Override
                    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<Tuple2<JSONObject, configTable>> out) throws Exception {
                        //主流数据来时，判断一下, 状态里面是否存在该表，存在的话数据是我们想要的，封装成一个元组抛出
                        //只读状态
                        ReadOnlyBroadcastState<String, configTable> state = ctx.getBroadcastState(broadcastDescriptor);

                        //数据中的表名
                        String table = value.getString("table");

                        //看状态中是否有该表
                        configTable configTable = state.get(table);

                        if (configTable!=null){

                            //数据中其他元数据没有用，所以取出data部分即可
                            JSONObject data = value.getJSONObject("data");

                            //补充一个字段 后续会用到  op_type
                             data.put("op_type", value.getString("type"));

                            out.collect(Tuple2.of(data,configTable));
                        }
                    }
                });



    }

    private SingleOutputStreamOperator<configTable> readConfigTable(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop162")
                .port(3306)
                .databaseList("edu_config") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("edu_config.table_process") // set captured table
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        return
                env
                        .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "table_process")
                        .map(json -> {
                            JSONObject jsonObject = JSON.parseObject(json);

                            return jsonObject.getObject("after", configTable.class);
                        });
    }

    private SingleOutputStreamOperator<JSONObject> etlData(DataStreamSource<String> stream) {

        return
                stream
                        .flatMap(new FlatMapFunction<String, JSONObject>() {
                            @Override
                            public void flatMap(String jsonStr, Collector<JSONObject> out) throws Exception {

                                try {
                                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                                    String database = jsonObj.getString("database");
                                    String type = jsonObj.getString("type");

                                    if ("edu".equals(database) && (
                                            "insert".equals(type)
                                                    || "update".equals(type)
                                                    ||"bootstrap-insert".equals(type))
                                            && jsonObj.getString("data") != null
                                            && jsonObj.getString("data").length() >2)
                                    {
                                        out.collect(jsonObj);
                                        System.out.println("etl数据完成"+ jsonObj);
                                    }
                                } catch (Exception e) {
                                    System.out.println("你的数据格式有问题");
                                }
                            }
                        });


    }

}
